package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //1.接收消息
        KStream<String, String> kStream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        kStream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(String key, String value) {
                        //1.接收消息
                        UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                        /**
                         *  2.重新定义key和value的值
                         *  key=文章id,value=类型：增量
                         */
                        return new KeyValue<String, String>(mess.getArticleId().toString(), mess.getType().name() + ":" + mess.getAdd());
                    }
                })
                //以key进行分组聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //聚合计算
                .aggregate(new Initializer<String>() {
                               /**
                                * 初始化聚合，返回值是消息的value
                                * 初始化，COLLECTION收藏分为0，COMMENT评论分为0，LIKES点赞分为0，VIEWS阅读分为0
                                */
                               @Override
                               public String apply() {
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                           }
                        //正式聚合计算
                        , new Aggregator<String, String, String>() {
                            /**
                             * 进行分值统计
                             * @param key key值  文章id
                             * @param value value值   类型:增量值 like:1
                             * @param init  初始化返回值， COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                             * @return
                             */
                            @Override
                            public String apply(String key, String value, String init) {
                                //如果value值为空
                                if (StringUtils.isEmpty(value)) {
                                    return init;//直接返回初始化值
                                }
                                //定义变量
                                int collection = 0, comment = 0, like = 0, view = 0;
                                //字符串转化成数组
                                String[] aggArray = init.split(",");//[COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0]
                                for (String agg : aggArray) {//COLLECTION:0
                                    String[] split = agg.split(":");
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                        case COLLECTION:
                                            collection = Integer.parseInt(split[1]);
                                            break;
                                        case COMMENT:
                                            comment = Integer.parseInt(split[1]);
                                            break;
                                        case LIKES:
                                            like = Integer.parseInt(split[1]);
                                            break;
                                        case VIEWS:
                                            view = Integer.parseInt(split[1]);
                                            break;
                                    }
                                }
                                /**
                                 * 累加操作，从value中取出分值，然后累加到变量中
                                 */
                                String[] split = value.split(":");
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                    case COLLECTION:
                                        collection += Integer.parseInt(split[1]);
                                        break;
                                    case COMMENT:
                                        comment += Integer.parseInt(split[1]);
                                        break;
                                    case LIKES:
                                        like += Integer.parseInt(split[1]);
                                        break;
                                    case VIEWS:
                                        view += Integer.parseInt(split[1]);
                                        break;
                                }

                                //返回类型：分值
                                String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);
                                System.out.println("文章的id:" + key);
                                System.out.println("当前时间窗口内的消息处理结果：" + formatStr);
                                return formatStr;
                            }
                        }, Materialized.as("hot-article_stream"))
                .toStream()
                .map((key, value) -> {
                    System.out.println("key值:" + key.key().toString());
                    System.out.println("value值：" + value.toString());

                    return new KeyValue<String, String>(key.key().toString(), formatMess(key.key().toString(), value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return kStream;
    }

    /**
     * 格式化value值，然后再发送消息
     * 把COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0格式化成ArticleVisitStreamMess类型
     *
     * @param key   文章id
     * @param value COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
     * @return
     */
    private String formatMess(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.parseLong(key));

        String[] aggArray = value.split(",");
        for (String agg : aggArray) {
            String[] split = agg.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        log.info("聚合消息处理之后的结果为:{}", JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}